package cn.itcast.czxy.BD18

import cn.itcast.czxy.BD18.bean.{BaseMode, Tagsfour}
import org.apache.spark.sql.{DataFrame, Dataset, Row}

object CustomerUnitPrice extends BaseMode{
  override def setAppName: String = "CustomerUnitPrice"

  override def setLeven4Id: Int = 126

  override def getNewTag(leve5: DataFrame, hbaseDF: DataFrame): DataFrame = {
    import spark.implicits._
    val leven5 = leve5.map(row => {
      val id = row.getAs("id").toString
      val rule = row.getAs("rule").toString.split("-")
      (id, rule(0).toInt, rule(1).toInt)
    }).toDF("id", "start", "stop")

    hbaseDF.createOrReplaceTempView("order")
    val orders: DataFrame = spark.sql("SELECT memberId,sum(productAmount)/COUNT(*) AS ddsl FROM order GROUP BY memberId")

    orders.join(leven5,orders("ddsl").between(leven5("start"),leven5("stop"))).select(orders("memberId").as("userId"),leven5("id").as("tagsId"))
  }


  def main(args: Array[String]): Unit = {
   exec()
  }
}
